package org.sg20.wordcooccurrence;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;

public class WordCooccurrenceMapper extends Mapper<LongWritable, Text, Text, Text> {
    HashMap<String, String> redhashmap = new HashMap<>();
    Text outKey = new Text();
    Text outVal = new Text();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException {

        Path rednameList = new Path(context.getCacheFiles()[0]);
        FileSystem fs = FileSystem.get(context.getConfiguration());
        InputStreamReader isr = new InputStreamReader(fs.open(rednameList));


        BufferedReader in = new BufferedReader(isr);

        // read until end of file
        String line;
        while ((line = in.readLine()) != null) {
            // add the name to the trie tree
            String[] names = line.split(" ");
            //Construct HashMap, the redundancy name is the key, and the real name is the value.
            for (int i = 1; i < names.length; i++) {
                redhashmap.put(names[i], names[0]);
            }
        }
        in.close();
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {

        String[] values = value.toString().split("\t");
        outKey.set(values[0]);

        HashMap<String, Integer> rmap= new HashMap<>();
        for (int i = 1; i < values.length; i++) {
            if (redhashmap.containsKey(values[i])) {
                if(!rmap.containsKey(redhashmap.get(values[i]))){
                    outVal.set(redhashmap.get(values[i]));
                    context.write(outKey, outVal);
                    rmap.put(redhashmap.get(values[i]), 1);
                }
            } else {
                if(!rmap.containsKey(values[i])) {
                    outVal.set(values[i]);
                    context.write(outKey, outVal);
                    rmap.put(values[i], 1);
                }
            }


        }

    }

}
